package com.sunday.common.mq.rocket.brave.study.apache.e1_Normal_Message;

import com.sunday.common.mq.rocket.brave.study.apache.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Stream;

/**
 * 发送消息
 * https://github.com/apache/rocketmq-spring/wiki/Send-Message
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class NormalMessageApi {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    private static final String topic = "your_topic";

    /**
     * ====================================== RocketMQTemplate ================================
     */

    /**
     * 同步发送消息
     */
    @GetMapping("syncSend")
    public void syncSend() {

        Map map = new HashMap() {{
            put("a", "b");
        }};

        log.info("rocketMQTemplate.send()");
        Message<String> message = new GenericMessage("rocketMQTemplate.send(message)", map);
        // 将默认目的地配置为在没有的发送方法中使用 目的地参数。如果未配置默认目的，则发送方法 没有目的地参数将在调用时引发异常。
        rocketMQTemplate.setDefaultDestination(topic);
        rocketMQTemplate.send(message);

        Message<String> message1 =
                MessageBuilder
                        .withPayload("MessageBuilder.withPayload")
                        .setHeader("a", "b")
                        .build();
        rocketMQTemplate.send(message1);

        User user = User.builder().userName("john").userAge(33).build();
        rocketMQTemplate.convertAndSend(user);

        rocketMQTemplate.convertAndSend(topic, user);

        rocketMQTemplate.convertAndSend(topic, user, map);

        rocketMQTemplate.convertAndSend(user, msg -> {
            MessageHeaderAccessor accessor = MessageHeaderAccessor.getMutableAccessor(msg);
            accessor.setHeader("test", "test");
            return MessageBuilder.createMessage(msg.getPayload(), accessor.getMessageHeaders());
        });

        rocketMQTemplate.convertAndSend(topic, user, map, msg -> msg);

    }

    /**
     * 异步发送消息
     */
    @GetMapping("asyncSend")
    public void asyncSend() {

        Map map = new HashMap() {{
            put("a", "b");
        }};

        Message<String> message =
                MessageBuilder
                        .withPayload("asyncSend")
                        .setHeader("a", "b")
                        .build();

        rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("OK {}", sendResult.getOffsetMsgId());
            }

            @Override
            public void onException(Throwable e) {
                log.info("Exception {}", e);
            }
        }, 1000);

        User user = User.builder().userName("john").userAge(33).build();
        rocketMQTemplate.asyncSend(topic, user, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("OK {}", sendResult.getOffsetMsgId());
            }

            @Override
            public void onException(Throwable e) {
                log.info("Exception {}", e);
            }
        });

        Collection<Message<User>> collection = Stream.iterate(1, i -> i < 10, i -> i + 1)
                .map(integer ->
                        MessageBuilder.withPayload(User.builder().userName("john" + integer).userAge(33 + integer).build()).build()
                )
                .toList();

        rocketMQTemplate.asyncSend(topic, collection, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("OK {}", sendResult.getOffsetMsgId());
            }

            @Override
            public void onException(Throwable e) {
                log.info("Exception {}", e);
            }
        });
    }

    /**
     * 有序发送消息
     */
    @GetMapping("syncSendOrderly")
    public void syncSendOrderly() {

        User user = User.builder().userName("john").userAge(33).build();

        rocketMQTemplate.syncSendOrderly(topic, user, user.getUserAge().toString());
        rocketMQTemplate.syncSendOrderly(topic, user, user.getUserAge().toString());
        rocketMQTemplate.syncSendOrderly(topic, user, user.getUserAge().toString());

    }


}